package subscribe

import (
	"bytes"
	"context"
	"crypto/md5"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/zeebo/errs"
	"go.uber.org/zap"
	"io"
	"net/http"
	"time"
)

// Payload 推送数据
type Payload interface {
	Topic() string
}

// Subscriber 推送对象
type Subscriber struct {
	Url string `json:"url"`
	Key string `json:"key"`
}

type Config struct {
	Timeout time.Duration `help:"推送超时时间" default:"30s"`
}

type Subscribe struct {
	config   *Config
	log      *zap.Logger
	deviceId string
}

func NewSubscribe(deviceId string, conf *Config, log *zap.Logger) *Subscribe {
	return &Subscribe{
		config:   conf,
		deviceId: deviceId,
		log:      log.Named("subscribe"),
	}
}

// Push 推送订阅消息
func (s *Subscribe) Push(ss []Subscriber, payload Payload) error {
	if len(ss) == 0 {
		return nil
	}
	topic := payload.Topic()
	data, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), s.config.Timeout)
	defer cancel()
	ers := errs.Group{}
	errChan := make(chan error, len(ss))
	for i := range ss {
		go func(sb Subscriber) {
			errChan <- s.push(ctx, sb, topic, data)
		}(ss[i])
	}
	for i := 0; i < len(ss); i++ {
		ers.Add(<-errChan)
	}
	return ers.Err()
}

func (s *Subscribe) push(ctx context.Context, ss Subscriber, topic string, data []byte) error {
	req, err := http.NewRequestWithContext(ctx, "POST", ss.Url, bytes.NewBuffer(data))
	if err != nil {
		return err
	}
	ts := fmt.Sprintf("%d", time.Now().Unix())
	//数据签名，md5(json数据+deviceId+topic+timestamp+key)
	signStr := append(data, []byte(s.deviceId)...)
	signStr = append(signStr, []byte(topic)...)
	signStr = append(signStr, []byte(ts)...)
	signStr = append(signStr, []byte(ss.Key)...)
	hash := md5.Sum(signStr)
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Content-Length", fmt.Sprintf("%d", len(data)))
	req.Header.Set("X-DEVICE-ID", s.deviceId)             //盒子id
	req.Header.Set("X-TOPIC", topic)                      //订阅主题
	req.Header.Set("X-HMAC", hex.EncodeToString(hash[:])) //签名
	req.Header.Set("X-TIMESTAMP", ts)                     //时间戳 秒
	client := http.Client{}
	resp, err := client.Do(req)
	if err != nil {
		s.log.Error("push request error", zap.Error(err))
		return err
	}
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		s.log.Error("push io.ReadAll error", zap.Error(err))
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		s.log.Error("push resp.StatusCode error", zap.Int("httpCode", resp.StatusCode), zap.ByteString("body", body))
		return errors.New("http code error")
	} else {
		s.log.Debug("push  finish", zap.ByteString("body", body))
	}
	return nil
}
